package org.rcy.framework.job.registry.zookeeper;

import com.dangdang.ddframe.job.reg.base.CoordinatorRegistryCenter;
import com.dangdang.ddframe.job.reg.exception.RegExceptionHandler;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperConfiguration;
import com.google.common.base.Charsets;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.ACLBackgroundPathAndBytesable;
import org.apache.curator.framework.api.ACLProvider;
import org.apache.curator.framework.api.transaction.CuratorTransactionBridge;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.data.ACL;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * @author 吴康桥
 * @Description
 * @date 2023/11/1 14:25
 */


public final class ZookeeperRegistryCenter implements CoordinatorRegistryCenter {

	private static final Logger logger = LoggerFactory.getLogger(ZookeeperRegistryCenter.class);
	private ZookeeperConfiguration zkConfig;
	private final Map<String, TreeCache> caches = new HashMap();
	private CuratorFramework client;

	public ZookeeperRegistryCenter(final ZookeeperConfiguration zkConfig) {
		this.zkConfig = zkConfig;
	}

	public void init() {
		logger.debug("Elastic job: zookeeper registry center init, server lists is: {}.", this.zkConfig.getServerLists());
		CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder().connectString(this.zkConfig.getServerLists()).retryPolicy(new ExponentialBackoffRetry(this.zkConfig.getBaseSleepTimeMilliseconds(), this.zkConfig.getMaxRetries(), this.zkConfig.getMaxSleepTimeMilliseconds())).namespace(this.zkConfig.getNamespace());
		if (0 != this.zkConfig.getSessionTimeoutMilliseconds()) {
			builder.sessionTimeoutMs(this.zkConfig.getSessionTimeoutMilliseconds());
		}

		if (0 != this.zkConfig.getConnectionTimeoutMilliseconds()) {
			builder.connectionTimeoutMs(this.zkConfig.getConnectionTimeoutMilliseconds());
		}

		if (!Strings.isNullOrEmpty(this.zkConfig.getDigest())) {
			builder.authorization("digest", this.zkConfig.getDigest().getBytes(Charsets.UTF_8)).aclProvider(new ACLProvider() {
				public List<ACL> getDefaultAcl() {
					return ZooDefs.Ids.CREATOR_ALL_ACL;
				}

				public List<ACL> getAclForPath(final String path) {
					return ZooDefs.Ids.CREATOR_ALL_ACL;
				}
			});
		}

		this.client = builder.build();
		this.client.start();

		try {
			if (!this.client.blockUntilConnected(this.zkConfig.getMaxSleepTimeMilliseconds() * this.zkConfig.getMaxRetries(), TimeUnit.MILLISECONDS)) {
				this.client.close();
				throw new KeeperException.OperationTimeoutException();
			}
		} catch (Exception var3) {
			RegExceptionHandler.handleException(var3);
		}

	}

	public void close() {
		Iterator var1 = this.caches.entrySet().iterator();

		while(var1.hasNext()) {
			Map.Entry<String, TreeCache> each = (Map.Entry)var1.next();
			((TreeCache)each.getValue()).close();
		}

		this.waitForCacheClose();
		CloseableUtils.closeQuietly(this.client);
	}

	private void waitForCacheClose() {
		try {
			Thread.sleep(500L);
		} catch (InterruptedException var2) {
			Thread.currentThread().interrupt();
		}

	}

	public String get(final String key) {
		TreeCache cache = this.findTreeCache(key);
		if (null == cache) {
			return this.getDirectly(key);
		} else {
			ChildData resultInCache = cache.getCurrentData(key);
			if (null != resultInCache) {
				return null == resultInCache.getData() ? null : new String(resultInCache.getData(), Charsets.UTF_8);
			} else {
				return this.getDirectly(key);
			}
		}
	}

	private TreeCache findTreeCache(final String key) {
		Iterator var2 = this.caches.entrySet().iterator();

		Map.Entry entry;
		do {
			if (!var2.hasNext()) {
				return null;
			}

			entry = (Map.Entry)var2.next();
		} while(!key.startsWith((String)entry.getKey()));

		return (TreeCache)entry.getValue();
	}

	public String getDirectly(final String key) {
		try {
			return new String((byte[])this.client.getData().forPath(key), Charsets.UTF_8);
		} catch (Exception var3) {
			RegExceptionHandler.handleException(var3);
			return null;
		}
	}

	public List<String> getChildrenKeys(final String key) {
		try {
			List<String> result = (List)this.client.getChildren().forPath(key);
			Collections.sort(result, new Comparator<String>() {
				public int compare(final String o1, final String o2) {
					return o2.compareTo(o1);
				}
			});
			return result;
		} catch (Exception var3) {
			RegExceptionHandler.handleException(var3);
			return Collections.emptyList();
		}
	}

	public int getNumChildren(final String key) {
		try {
			Stat stat = (Stat)this.client.checkExists().forPath(key);
			if (null != stat) {
				return stat.getNumChildren();
			}
		} catch (Exception var3) {
			RegExceptionHandler.handleException(var3);
		}

		return 0;
	}

	public boolean isExisted(final String key) {
		try {
			return null != this.client.checkExists().forPath(key);
		} catch (Exception var3) {
			RegExceptionHandler.handleException(var3);
			return false;
		}
	}

	public void persist(final String key, final String value) {
		try {
			if (!this.isExisted(key)) {
				((ACLBackgroundPathAndBytesable)this.client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT)).forPath(key, value.getBytes(Charsets.UTF_8));
			} else {
				this.update(key, value);
			}
		} catch (Exception var4) {
			RegExceptionHandler.handleException(var4);
		}

	}

	public void update(final String key, final String value) {
		try {
			((CuratorTransactionBridge)((CuratorTransactionBridge)this.client.inTransaction().check().forPath(key)).and().setData().forPath(key, value.getBytes(Charsets.UTF_8))).and().commit();
		} catch (Exception var4) {
			RegExceptionHandler.handleException(var4);
		}

	}

	public void persistEphemeral(final String key, final String value) {
		try {
			if (this.isExisted(key)) {
				this.client.delete().deletingChildrenIfNeeded().forPath(key);
			}

			((ACLBackgroundPathAndBytesable)this.client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL)).forPath(key, value.getBytes(Charsets.UTF_8));
		} catch (Exception var4) {
			RegExceptionHandler.handleException(var4);
		}

	}

	public String persistSequential(final String key, final String value) {
		try {
			return (String)((ACLBackgroundPathAndBytesable)this.client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT_SEQUENTIAL)).forPath(key, value.getBytes(Charsets.UTF_8));
		} catch (Exception var4) {
			RegExceptionHandler.handleException(var4);
			return null;
		}
	}

	public void persistEphemeralSequential(final String key) {
		try {
			((ACLBackgroundPathAndBytesable)this.client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL_SEQUENTIAL)).forPath(key);
		} catch (Exception var3) {
			RegExceptionHandler.handleException(var3);
		}

	}

	public void remove(final String key) {
		try {
			this.client.delete().deletingChildrenIfNeeded().forPath(key);
		} catch (Exception var3) {
			RegExceptionHandler.handleException(var3);
		}

	}

	public long getRegistryCenterTime(final String key) {
		long result = 0L;

		try {
			this.persist(key, "");
			result = ((Stat)this.client.checkExists().forPath(key)).getMtime();
		} catch (Exception var5) {
			RegExceptionHandler.handleException(var5);
		}

		Preconditions.checkState(0L != result, "Cannot get registry center time.");
		return result;
	}

	public Object getRawClient() {
		return this.client;
	}

	public void addCacheData(final String cachePath) {
		TreeCache cache = new TreeCache(this.client, cachePath);

		try {
			cache.start();
		} catch (Exception var4) {
			RegExceptionHandler.handleException(var4);
		}

		this.caches.put(cachePath + "/", cache);
	}

	public void evictCacheData(final String cachePath) {
		TreeCache cache = (TreeCache)this.caches.remove(cachePath + "/");
		if (null != cache) {
			cache.close();
		}

	}

	public Object getRawCache(final String cachePath) {
		return this.caches.get(cachePath + "/");
	}

	public ZookeeperConfiguration getZkConfig() {
		return this.zkConfig;
	}

	public void setZkConfig(ZookeeperConfiguration zkConfig) {
		this.zkConfig = zkConfig;
	}

	public CuratorFramework getClient() {
		return this.client;
	}

	public void setClient(CuratorFramework client) {
		this.client = client;
	}
}
